package cgl.hadoopsensorgrid.sensorgrid.rtdtohadoop;

import java.lang.InterruptedException;
import java.lang.StringBuffer;
import java.io.IOException;
import java.io.File;
import java.io.FileWriter;
import java.io.ObjectOutputStream;
import java.nio.ByteBuffer;
import java.nio.channels.SocketChannel;
import java.nio.channels.Selector;
import java.nio.channels.SelectionKey;
import java.net.InetSocketAddress;
import java.net.Socket;
import java.net.ServerSocket;
import java.net.InetAddress;
import java.util.GregorianCalendar;
import java.util.Calendar;
import java.util.StringTokenizer;
import java.util.Iterator;
import java.util.Properties;

import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.io.BytesWritable;
import org.apache.hadoop.io.Text;

import cgl.hadoopsensorgrid.sensorgrid.util.executeCommand;
import cgl.hadoopsensorgrid.sensorgrid.util.PropertyFile;
import cgl.hadoopsensorgrid.sensorgrid.util.testingLog;

/***
 * 
 * The rtdMapper extends the hadoop Mapper class. It is deployed by the
 * rtdController. It is used for getting data from the GPS networks and passing
 * it to the rtdReduce.
 * 
 * @author Chao Sun
 * 
 */
public class rtdMapper extends Mapper<Object, Text, Text, BytesWritable> {
	private Calendar cal = null;
	private int TIMEOUT = 50;
	private Properties properties;
	private String hdfsScript;
	// variables for RTD
	private String rtdAddr;
	private int rtdPort;
	private String rtdAddress;
	private InetSocketAddress rtdSockAddr = null;
	private SocketChannel socketClient = null;
	private Selector sel = null;
	// variables for RYO
	private String ryoAddr = null;
	private int ryoPort = 5000;
	private ServerSocket ryoSocket = null;
	private Socket openedSocket = null;
	private ObjectOutputStream ryoOutStream = null;
	private String ryoFileBase;
	private String ryoInputBase;

	/***
	 * 
	 * The class extends class Thread for opening a socket and listening on it.
	 * 
	 * @author Chao Sun
	 * 
	 */
	class openRyoThread extends Thread {

		public openRyoThread() {
		}

		/**
		 * Open and listen on a socket.
		 */
		public void run() {
			// Try opening
			try {
				ryoSocket = new ServerSocket(ryoPort);
			} catch (IOException ioE) {
				System.out.println(ryoAddr
						+ " RYO Socket opening unsuccessful. Terminating...");
				System.out.println(ioE.toString());
				System.exit(0);
			}
			System.out.println(ryoAddr + " RYO socket opened successfully");
			// Try listening
			try {
				openedSocket = ryoSocket.accept();
				ryoOutStream = new ObjectOutputStream(openedSocket.getOutputStream());
			} catch (IOException ioE) {
				System.out.println(ryoAddr
						+ " RYO Socket listening error. Terminating...");
				System.out.println(ioE.toString());
				System.exit(0);
			}
			System.out.println("Connection recieved from "
					+ openedSocket.getInetAddress().getHostAddress());
		}

	}

	/**
	 * This method connects to an RTD socket address.
	 * 
	 * @throws IOException
	 */
	private void connectRtdSocket() throws IOException {
		socketClient = SocketChannel.open();
		socketClient.connect(rtdSockAddr);
		socketClient.configureBlocking(false);
		socketClient.register(sel, SelectionKey.OP_READ);
	}

	/**
	 * This method starts a connection to a remote GPS network, and gets RYO data
	 * therefrom. The data is then passed to RYO data structure via a socket. Time
	 * out or server error may occur which require the socket to be closed and
	 * reconnected.
	 * 
	 * Code from cgl.sensorgrid.sopac.gps.filter.Rtd2Narada.java by Galip Aydin.
	 */
	private void run() {
		byte[] buffer = null;
		int timeout = 0;
		// Get and pass data. Deal with exceptions.
		while (true) {
			try {
				int n = sel.select(300000); // time out setting is 5 minutes
				if (n == 0) {
					socketClient.close();
					connectRtdSocket();
					System.out.println("Time out. " + rtdAddress + " reconnected");
				}

				Iterator it = sel.selectedKeys().iterator();
				while (it.hasNext()) {
					SelectionKey key = (SelectionKey) it.next();
					if (key.isReadable()) {
						ByteBuffer buf = ByteBuffer.allocate(2048);
						byte[] by;
						int count = socketClient.read(buf);
						if (count > 0) {
							buf.flip();
							by = new byte[count];
							if (buf.array().length > 0) {
								System.arraycopy(buf.array(), 0, by, 0, count);
								int len = by.length;
								if (by != null && len > 0) { // good to send
									timeout++;

									// RYO timeout
									if (timeout >= TIMEOUT) {
										timeout = 0;
										System.out.println("Client timeout");
										// send terminate signal
										if (openedSocket != null && ryoOutStream != null) {
											String terminateSignal = new String("Terminate");
											ryoOutStream.writeObject(terminateSignal.getBytes());
											ryoOutStream.flush();
											// System.out.println("Send terminating message");
											buffer = by;
											// System.out.println("Store the first RYO data set");
											try {
												ryoSocket.close();
												openedSocket = null;
												ryoOutStream = null;
												openRyoThread openThread = new openRyoThread();
												openThread.start();
											} catch (Exception e) {
												System.out.println(ryoAddr + " RYO socket open error");
											}
										} else { // timeout before connection
											testingLog.log(ryoFileBase + File.separator,
													"RYO connection failure. Exit\n");
											System.out.println("RYO connection failure. Exit");
											System.exit(0);
										}
									}

									// connection setup, send data
									if (openedSocket != null && ryoOutStream != null) {
										try {
											if (buffer != null) { // send stored data if there is any
												ryoOutStream.writeObject(buffer);
												buffer = null;
												// System.out.println("Send stored RYO data");
											}
											ryoOutStream.writeObject(by);
											ryoOutStream.flush();
											// System.out.println("Send RYO data");
										} catch (IOException ioE) {
											System.out.println("Client disconnected");
											System.exit(0);
										}
									} else if (buffer == null) { // no connection no buffered data
										buffer = by;
										// System.out.println("Store the very first RYO data set");
									} else { // no connection, append buffered data
										int bufferLen = buffer.length;
										int byLen = by.length;
										int i;
										int j;
										byte[] temp = new byte[bufferLen + byLen];
										for (i = 0; i < bufferLen; i++) {
											temp[i] = buffer[i];
										}
										for (j = 0; j < byLen; i++, j++) {
											temp[i] = by[j];
										}
										buffer = temp;
										testingLog.log(ryoFileBase + File.separator, "Store RYO data");
									}

								} // if

								count = 0;
								buf.clear();
							} else {
								System.err.println(rtdAddress + ": NULL RYO data");
							}
						} else if (count == -1) {
							buf = null;
							socketClient.close();
							connectRtdSocket();
							System.out.println("End of stream or socket " + rtdAddress
									+ " closed by server. Reconnected");
						}
						buf = null;
					} // if

					it.remove();
					cal = new GregorianCalendar();
				} // while

			} catch (IOException ioE) {
				try {
					if (socketClient.isOpen() || socketClient.isConnected()) {
						socketClient.close();
					}
					connectRtdSocket();
					System.out.println("Server error. " + rtdAddress + " reconnected");
				} catch (IOException cE) {
					try {
						socketClient.close();
					} catch (IOException e) {
						e.printStackTrace();
					}
				}
			} // try

		} // while

	}

	/**
	 * This method starts a thread for opening and listening on a socket, through
	 * which the RYO data is sent out. It also calls the run() method to get RYO
	 * data.
	 * 
	 * @param context
	 *          context of this rtdMapper object
	 * 
	 * @throws InterruptedException
	 */
	private void runRtdClient(Context context) throws InterruptedException {
		cal = new GregorianCalendar();

		// Try connecting
		try {
			connectRtdSocket();
		} catch (IOException ex) {
			System.out.println(rtdAddress
					+ " RTD Socket connection unsuccessful. Terminating...");
			System.out.println(ex.toString());
			System.exit(0);
		}
		// Try opening
		System.out.println(rtdAddress + " RTD socket connection successful at "
				+ cal.get(Calendar.HOUR_OF_DAY) + ":" + cal.get(Calendar.MINUTE) + ":"
				+ cal.get(Calendar.SECOND));

		openRyoThread openThread = new openRyoThread();
		openThread.start();
		run();
	}

	/**
	 * Override the setup method.
	 * 
	 * @param context
	 *          context of this rtdMapper object
	 */
	protected void setup(Context context) {
		loadProperties();
		try {
		testingLog.log(ryoFileBase + File.separator, "Rtd mapper started\n");
		} catch (IOException e) {
		}
	}

	/**
	 * This method overrides the map() method of class Mapper. It gets the GPS
	 * network address and port number from a file and assigns them to the two
	 * global variables, and then starts the runRtdClient.
	 * 
	 * @param key
	 *          key of the data split
	 * @param value
	 *          contents of the data split, which is the address and port
	 * @param context
	 *          context of this rtdMapper object
	 * 
	 * @throws InterruptedException
	 * 
	 * @throws IOException
	 */
	protected void map(Object key, Text value, Context context)
			throws IOException, InterruptedException {
		StringTokenizer sTokenizer = new StringTokenizer(value.toString(), ":");
		int tokenNum = sTokenizer.countTokens();

		if (tokenNum != 2) {
			System.err
					.println("Network list error. Format: " + "<Network IP>:<Port>");
			System.exit(1);
		}

		rtdAddr = sTokenizer.nextToken();
		rtdPort = Integer.parseInt(sTokenizer.nextToken());
		rtdSockAddr = new InetSocketAddress(rtdAddr, rtdPort);
		sel = Selector.open();
		rtdAddress = rtdAddr + ":" + rtdPort;

		// write RYO socket address and port to file
		ryoAddr = InetAddress.getLocalHost().getHostAddress();
		String ryoFileSourse = ryoFileBase + File.separator + ryoAddr + "_ryo.txt";
		FileWriter filewriter = new FileWriter(ryoFileSourse);
		filewriter.write(ryoAddr + ":" + ryoPort + "\n");
		filewriter.flush();
		filewriter.close();
		testingLog.log(ryoFileBase + File.separator, "Created local ryo input file\n");

		// put source file to HDFS
		testingLog.log(ryoFileBase + File.separator, "Putting RYO socket file to HDFS\n");
		String command = hdfsScript + " -put " + ryoFileSourse + " " + ryoInputBase
				+ ryoAddr + "_ryo.txt";
		executeCommand.execute(command);
		testingLog.log(ryoFileBase + File.separator, "Putted file to HDFS. Sleeping\n");
		Thread.sleep(3000);
		testingLog.log(ryoFileBase + File.separator, "Waked up. Run RTD sockets\n");

		runRtdClient(context);
	}

	/**
	 * Load properties from a given file.
	 */
	private void loadProperties() {
		try {
			properties = PropertyFile.loadProperties("hadoopsensorgrid.properties");
			hdfsScript = properties.getProperty("hdfs.hadoop.script");
			ryoFileBase = properties.getProperty("hadoop.ryo.inputbase");
			ryoInputBase = properties.getProperty("hdfs.ryo.inputbase");
			if (properties == null || ryoInputBase == null || hdfsScript == null
					|| ryoFileBase == null) {
				System.err.println("Error: rtdMapper failed to load properties!");
				System.exit(-1);
			}
		} catch (Exception e) {
			e.printStackTrace();
		}
	}

}